package com.sojson.util.mq.rabbit;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.GetResponse;
import com.rabbitmq.client.Return;
import com.rabbitmq.client.ReturnCallback;
import com.rabbitmq.client.ReturnListener;
import com.sojson.constant.Constant;

/**
 * Rabbit消息队列工具类
 * 
 * @author liu
 * @date 2021-07-11
 */
public class RabbitUtil {

    /**
     * BuiltinExchangeType.FANOUT</BR>
     * 将消息分发到所有的绑定队列,不处理路由键
     * 你只需要简单的将队列绑定到交换机上
     * 一个发送到该类型交换机的消息都会被广播到与该交换机绑定的所有队列上
     * 
     * BuiltinExchangeType.DIRECT</BR>
     * 按照routingkey分发到指定队列,处理路由键，需要将一个队列绑定到交换机上
     * 要求该消息与一个特定的路由键完全匹配,这是一个完整的匹配
     * 如果一个队列绑定到该交换机上要求路由键为 “green”,则只有路由键为“green”的消息才被转发
     * 不会转发路由键为“red”,只会转发路由键为“green”
     * 
     * BuiltinExchangeType.TOPIC</BR>
     * 模糊匹配,将路由键和某模式进行匹配,此时队列需要绑定要一个模式上
     * 符号“#”匹配一个或多个词,符号“*”只能匹配一个词
     * 
     * BuiltinExchangeType.HEADERS</BR>
     * 通过添加属性key-value匹配,不处理路由键,而是根据发送的消息内容中的headers属性进行匹配
     * 在绑定Queue与Exchange时指定一组键值对
     * 当消息发送到RabbitMQ时会取到该消息的headers与Exchange绑定时指定的键值对进行匹配
     * 如果完全匹配则消息会路由到该队列,否则不会路由到该队列
     * headers属性是一个键值对,可以是Hashtable
     * 键值对的值可以是任何类型,而fanout,direct,topic 的路由键都需要要字符串形式的
     * @throws Exception 
     */
    public static void main(String[] args) throws Exception {
        a1("directExchange1", "test1", "test1消息");
        a1("directExchange1", "test2", "test2消息");
        // b1("directExchange1", "rabbitmq1", "test1a");
    }

    public static void a1(String direct, String routingKey, String content) throws Exception {
        Channel channel = ChannelUtil.getChannelInstance("测试");

        // 声明交换机和队列 然后进行绑定设置, 最后制定路由Key
        // 创建交换机
        /**
         * exchange:    交换机名称
         * type:        交换机类型,共四个direct,fanout,topic,headers
         * durable:     消息是否持久化(默认为false)
         * autoDelete:  当交换机不再使用时,是否自动删除它(默认为false)
         * internal:    设置是否内置的,如果设置为true,则表示是内置的交换器(默认为false)
         *              客户端程序无法直接发送消息到这个交换器中,只能通过交换器路由到交换器这种方式
         * arguments:   其他参数(默认为null)
         */
        channel.exchangeDeclare(direct, BuiltinExchangeType.DIRECT, true);
        /**
         * 将通道与队列绑定
         * queue:       绑定的队列名,没有则创建
         * durable:     是否持久化
         *              队列默认是放在内存中的,rabbitmq服务挂了队列被删除
         *              设置为true则持久化到磁盘,rabbitmq服务挂了重启还存在
         * exclusive:   是否独占,设置为true表示独占,一般用于一个生产者和一个消费者场景,当连接关闭时会删除该队列
         * autoDelete:  是否自动删除,自动删除前提是之前有消费者,当消费者全部断开连接后,则自动删除
         * arguments:   其他参数(默认为null)
         */
        // channel.queueDeclare("", false, false, false, null);
        // private String contentType; // 消息的内容类型
        // private String contentEncoding; // 消息内容的编码
        // private Map<String, Object> headers; // 消息的头
        // private Integer deliveryMode; // 消息是否持久化，1为否，2为是
        // private Integer priority; // 消息的优先级
        // private String correlationId; // 关联id
        // private String replyTo; // 消息回复的队列名称
        // private String expiration; // 消息的有效时间(单位: 毫秒)
        // private String messageId; // 消息的id
        // private Date timestamp; // 消息的时间戳
        // private String type; // 类型
        // private String userId; // 用户id
        // private String appId; // 应用id
        // private String clusterId; // 集群id
        AMQP.BasicProperties basicProperties =
            new AMQP.BasicProperties().builder().deliveryMode(2).contentType(Constant.DEFAULT_ENCODING).build();
        // 发送消息
        content = "交换机: " + direct + ", Key: " + routingKey + "发送消息: " + content;
        System.out.println(content);
        /**
         * 消息发送成功的异步回调
         */
        channel.addConfirmListener(new ConfirmListener() {

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                // TODO Auto-generated method stub

            }

            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                // TODO Auto-generated method stub

            }
        });
        channel.addReturnListener(new ReturnCallback() {
            @Override
            public void handle(Return returnMessage) {
                System.out.println("消息发送失败的同步回调");
            }
        });
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int arg0, String arg1, String arg2, String arg3, BasicProperties arg4, byte[] arg5)
                throws IOException {
                System.out.println("消息发送失败的异步回调");
            }
        });
        // 发送消息
        /**
         * exchange:    交换机的名称
         * routingKey:  消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * mandatory:   是否开启消息故障检测模式
         *              它只会让RabbitMq向你通知失败,而不会通知成功
         *              设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *              channel.addReturnListener(new ReturnListener(){})为异步返回
         *              如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *              如果为false,则broker会直接将消息丢弃
         * immediate:   RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *              原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *              当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *              immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *              如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * props:       其它的一些属性(内容格式,类型,请求头等)
         * body:        内容
         */
        channel.basicPublish(direct, routingKey, false, basicProperties, "body中的消息内容！".getBytes());
        channel.basicPublish(content, routingKey, false, false, basicProperties, null);
        channel.basicPublish(direct, routingKey, false, null, content.getBytes());
        channel.basicPublish(direct, routingKey, false, null, content.getBytes());
        channel.basicPublish(direct, routingKey, false, null, content.getBytes());
        channel.basicPublish(direct, routingKey, false, null, content.getBytes());
        channel.basicPublish(direct, routingKey, false, null, content.getBytes());
        // 关闭连接,一般消费者不关闭,一直监听队列消息并消费
        ChannelUtil.close(channel, null);
    }

    /**
     * 
     * 
     * @param direct        交换机的名称
     * @param declareName   队列的名称
     * @param routingKey    key
     * @throws IOException
     * @throws TimeoutException
     */
    public static void b1(String direct, String declareName, String routingKey) throws IOException, TimeoutException {
        Channel channel = ChannelUtil.getChannelInstance("测试");

        // 声明交换机和队列 然后进行绑定设置, 最后制定路由Key
        // 创建交换机
        /**
         * exchange:    交换机名称
         * type:        交换机类型,共四个direct,fanout,topic,headers
         * durable:     消息是否持久化(默认为false)
         * autoDelete:  当交换机不再使用时,是否自动删除它(默认为false)
         * internal:    设置是否内置的,如果设置为true,则表示是内置的交换器(默认为false)
         *              客户端程序无法直接发送消息到这个交换器中,只能通过交换器路由到交换器这种方式
         * arguments:   其他参数(默认为null)
         */
        channel.exchangeDeclare(direct, BuiltinExchangeType.FANOUT, true, false, false, new HashMap<>());
        Map<String, Object> map = new HashMap<>();
        // 当x-max-length与x-max-length-bytes都设置时,无论哪个达到限制都会触发溢出策略
        // 溢出策略有两种,1:丢弃最前面的消息(默认),2:拒绝发布,修改方式为启动发布者确认机制(https://www.rabbitmq.com/confirms.html#publisher-confirms)
        // 队列最多容纳10条数据
        map.put("x-max-length", 10);
        // 消息最大长度
        map.put("x-max-length-bytes", 10);
        // 为当前队列设置存活时间(单位: 毫秒)
        map.put("x-expires", 1800000);
        // 为当前队列的每一条消息设置存活时间(单位: 毫秒)
        map.put("x-message-ttl", 1000);
        // 设置队列的优先级,参数可以写1到255之间,数字越大,优先级越高,实际使用中建议使用1到10之间选择,未设置的都为0
        // 设置优先级都有一些内存和磁盘成本,还有额外的CPU成本,优先级设置的越高消耗越大(erlang进程开启的越多)
        map.put("x-max-priority", 10);
        // 当存在多个消费者时,默认会使用循环的方式去消费消息
        // 当设置某个消费者优先级后,只有当改消费者受阻时消息才发送给低优先级消费者消费,同级别的优先级消费者仍然采取循环的方式消费
        map.put("x-priority", 1);
        // 声明队列,将通道与队列绑定
        /**
         * 将通道与队列绑定
         * queue:       绑定的队列名,没有则创建
         * durable:     是否持久化
         *              队列默认是放在内存中的,rabbitmq服务挂了队列被删除
         *              设置为true则持久化到磁盘,rabbitmq服务挂了重启还存在
         * exclusive:   是否独占,设置为true表示独占,一般用于一个生产者和一个消费者场景,当连接关闭时会删除该队列
         * autoDelete:  是否自动删除,自动删除前提是之前有消费者,当消费者全部断开连接后,则自动删除
         * arguments:   其他参数(默认为null)
         */
        channel.queueDeclare(declareName, true, false, false, map);
        /**
         * 将交换器与队列的路由键绑定
         * queue:       队列名称
         * exchange:    交换机名称
         * routingKey:  路由键
         * arguments:   其他参数(默认为null)
         */
        channel.queueBind(declareName, direct, routingKey, new HashMap<>());
        // 获取一个临时队列,该队列特点: 不持久化,排他,自动删除,也可以自己声明一个
        String queueName = channel.queueDeclare().getQueue();
        // 创建消费者
        Consumer consumer = new DefaultConsumer(channel) {

            /**
             * 收到消息后的处理
             * consumerTag: 
             * envelope:    
             * properties:  
             * body:        内容
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {

                String message = new String(body, "UTF-8");

                System.out.println(consumerTag);
                System.out.println(envelope);
                System.out
                    .println("交换机: " + direct + ", 队列: " + declareName + ", Key: " + routingKey + ",收到消息: " + message);

                try {
                    System.out.println(new String(body) + "----" + envelope.getDeliveryTag());
                    // 在处理完消息后手动进行确认
                    /*
                    * 参数1： 消费消息的index
                    * 参数2： 是否批量确认消息,前提是在同一个channel里面,且是在该消息确认前没有被确认的消息才能批量确认
                    * */
                    channel.basicAck(envelope.getDeliveryTag(), false);
                } catch (Exception e) {
                    /**
                     * deliveryTag: 消费消息的index
                     * requeue:     对异常消息的处理,true表示重排序,false表示丢弃
                     */
                    channel.basicReject(envelope.getDeliveryTag(), false);
                    /**
                     * deliveryTag: 消费消息的index
                     * multiple:    是否批量否定多个消息,设为false就与basicReject功能一样,true的前提也是在同一个channel,且在该消息否定前存在未确认的消息
                     * requeue:     对异常消息的处理,true表示重排序,false表示丢弃
                     */
                    channel.basicNack(envelope.getDeliveryTag(), false, true);
                }
            }

        };
        /**
         * 预取设置,允许该通道未确认交付的最大数量
         * 设置通道的预取数量为1，官方推荐100到300，数据会影响其吞吐量
         */
        channel.basicQos(4);
        // 开始消费消息
        /**
         * queue:       队列名
         * autoAck:     是否开启自动确认机制
         *              当rabbitmq将消息发送给消费者进行消费时,消费者可能会消费消息失败的情况,用户可以设置消费失败的消息给其他消费者消费或者直接丢弃
         *              自动确认模式积极的一面是能够拥有更高的吞吐量,但是却存在数据安全的问题
         * consumerTag: 
         * callback:    消息的回调接口,创建Consumer接口的实现类DefaultConsumer,重写handleDelivery方法,去处理队列中的数据
         */
        // 该模式的主体是队列,采用的是发布订阅的模式
        // 当队列中有新的消息时,队列就会主动将消息推送给消费者进行消费
        // 它保证了消费者能够及时的获取到相关的消息,其建立道德连接是长连接,比较节约资源,官方推荐使用该方法
        String basicConsume = channel.basicConsume(declareName, true, "消费者标签", consumer);
        /**
         * queue:       队列名
         * autoAck:     是否开启自动确认机制
         *              当rabbitmq将消息发送给消费者进行消费时,消费者可能会消费消息失败的情况,用户可以设置消费失败的消息给其他消费者消费或者直接丢弃
         *              自动确认模式积极的一面是能够拥有更高的吞吐量,但是却存在数据安全的问题
         */
        // 该模式的主体是消费者,当消费者需要队列中的一条消息时,可以通过该方法去获取到队列中的一条消息
        // 其建立的连接是短连接,当消费者获取到消息后该连接断开
        // 可以通过循环的方式实现和basicConsume相同的效果,但不建议这样去使用
        GetResponse work = channel.basicGet("work", true);
        System.out.println(new String(work.getBody()));
    }

}